package com.saibowisdom.storage.aspirator.es;

import com.alibaba.fastjson.JSONArray;
import com.saibowisdom.storage.aspirator.common.MetaRetrieve;
import com.saibowisdom.storage.aspirator.common.meta.NodeInfo;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import org.apache.commons.cli.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by admin on 2017/1/11.
 */
public class ElasticSearchReceiver {


    final static Logger logger = LoggerFactory.getLogger(ElasticSearchReceiver.class);

    static ESAccessor esDao = new ESAccessor();
    static MetaRetrieve mr = new MetaRetrieve();

    /**
     * 消费者配置
     *
     * @param zk zookeeper服务
     * @return ConsumerConfig
     */
    private static ConsumerConfig createConsumerConfig(String zk) {
        final Properties props = new Properties();
        props.put("zookeeper.connect", zk);
        props.put("group.id", "aspirator1");
        //props.put("group.id", "internetdbstore");
        props.put("zookeeper.session.timeout.ms", "4000");
        props.put("zookeeper.sync.time.ms", "200");
        props.put("auto.commit.interval.ms", "1000");
        props.put("auto.offset.reset", "smallest");

        return new ConsumerConfig(props);

    }


    public static void main(String[] args) {
        /** 获取本机ip和hostName */
        final InetAddress add;
        String ip = null;
        String hostName = null;
        try {
            add = InetAddress.getLocalHost();
            if (add != null) {
                ip = add.getHostAddress();
                hostName = add.getHostName();
            }
        } catch (UnknownHostException e) {
            System.out.println("Failed to get the IP and hostName---获取ip和hostName失败");
            return;
        }

        final CommandLineParser parser = new BasicParser();
        final Options options = new Options();
        options.addOption("z", "zookeeperlist", true, "zookeeper address list");
        options.addOption("e", "elasticsearch", true, "elasticsearch host");
        options.addOption("c", "escluster", true, "elasticsearch cluster name");
        options.addOption("j", "jdbc", true, "mysql jdbc url");
        options.addOption("u", "user", true, "mysql user");
        options.addOption("p", "passwd", true, "mysql password");
        // Parse the program arguments
        final CommandLine commandLine;
        try {
            commandLine = parser.parse(options, args);
        } catch (ParseException e) {
            e.printStackTrace();
            System.out.println("Parse command fails---解析输入命令失败");
            return;
        }
        final ConsumerConnector consumer;
        final String topic = "aspDataList";

        final String zk;
        if (commandLine.hasOption('z')) {
            zk = commandLine.getOptionValue('z');
        } else {
            System.out.println("ERROR:Zookeeper address not list");
            return;
        }
        consumer = kafka.consumer.Consumer.createJavaConsumerConnector(createConsumerConfig(zk));


        /**
         * 成功
         */
        final AtomicInteger number = new AtomicInteger(0);
        /**
         * 失败
         */
        final AtomicInteger error = new AtomicInteger(0);
        /**
         * 忽略
         */
        final AtomicInteger ignore = new AtomicInteger(0);
        /**
         * 是否有数据
         */
        final AtomicInteger stopflag = new AtomicInteger(0);

        final String es;

        if (commandLine.hasOption('e')) {
            es = commandLine.getOptionValue('e');
        } else {
            System.out.println("ERROR:elasticsearch host  not list");
            return;
        }

        final String jdbc;
        String user = null;
        String passwd = null;
        if (commandLine.hasOption('j')) {
            jdbc = commandLine.getOptionValue('j');
        } else {
            System.out.println("ERROR:mysql jdbc url  not list");
            return;
        }

        if (commandLine.hasOption('u')) {
            user = commandLine.getOptionValue('u');
        }

        if (commandLine.hasOption('p')) {
            passwd = commandLine.getOptionValue('p');
        }

        mr.setDataSource(jdbc, user, passwd);

        String cluster = null;
        if (commandLine.hasOption('c')) {
            cluster = commandLine.getOptionValue('c');
        }

        esDao.init(es, cluster);

        mr.getCache("");
        esDao.setParameter(mr);


        /** 更新节点状态 */
        final NodeInfo nodeInfo = new NodeInfo();
        nodeInfo.setIp(ip);
        nodeInfo.setHost(hostName);
        nodeInfo.setState(true);
        nodeInfo.setStop(false);
        nodeInfo.setType(1);
        //nodeInfo.setLocation(location);
        try {
            mr.updateNodeState(nodeInfo);
        } catch (Exception e) {
            System.out.println("Update the node status failed---更新节点状态失败");
            return;
        }


        final Map<String, Integer> topicCountMap = new HashMap<>();
        final int num = 1;
        topicCountMap.put(topic, num);

        //final Map<String, Integer> tableExsitMap = new HashMap<>();
        final ExecutorService executor = Executors.newFixedThreadPool(num);
        final Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);


        /**
         * 更新节点信息，判断节点是否存活
         */
        new Thread(new Runnable() {
            public void run() {
                while (true) {
                    try {
                        Thread.sleep(1000 * 30);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    try {
                        mr.updateNodeTime(nodeInfo);
                        //logger.info("节点状态已更新...时间 {} ", new Date().getTime());
                    } catch (Exception e) {
                        logger.error("更新 {} 时间失败,mysql插入失败 {}", nodeInfo.getIp(), e.getMessage());
                    }

                }
            }
        }).start();


        /**
         * 更新节点处理数据量
         */
        new Thread(new Runnable() {
            public void run() {
                while (true) {
                    try {
                        Thread.sleep(1000 * 20);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    try {
                        nodeInfo.setScount(number.get());
                        nodeInfo.setFcount(error.get());
                        nodeInfo.setIcount(ignore.get() > 0 ? ignore.get() : 0);
                        mr.updateNodeProcessNum(nodeInfo);
                        //logger.info("节点处理数量已更新 当前成功 {} ,忽略 {} ,失败 {}", number, ignore.get() > 0 ? ignore.get() : 0, error);
                    } catch (Exception e) {
                        logger.error("更新 {} 处理数量失败,mysql插入失败 {}", nodeInfo.getIp(), e.getMessage());
                    }

                }
            }
        }).start();

        /**
         * 重新插入错误为类型2的数据
         */
        /*new Thread(new Runnable() {

            public void run() {
                while (true){
                    try {
                        Thread.sleep(1000 * 60 * 15);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }

                    try {
                        final ArrayList<MonitorInfo> monitorInfos = esDao.getErrorData(nodeInfo);
                        logger.info("重新录入错误数据,共 {} 条", monitorInfos.size());

                        for (MonitorInfo monitorInfo : monitorInfos) {
                            stopflag.set(1);
                            final Object map12;
                            final String json = monitorInfo.getContent();
                            map12 = JSONArray.parse(json);

                            @SuppressWarnings("unchecked")
                            final Map<String, Object> map1 = (Map<String, Object>) map12;
                            final Object lst22 = map1.get("DATA");

                            @SuppressWarnings("unchecked")
                            final List<Map<String, String>> lst2 = (List<Map<String, String>>) lst22;

                            final String table = (String) map1.get("TABLE");

                            *//** 主键 *//*
                            final List<String> pkStr = mr.getPrikeyList(table);
                            for (Map<String, String> map2 : lst2) {
                                *//** 检查表结构 *//*
                                if (mr.checkSchema(table, map2)) {
                                    *//** 表结构有变化 *//*
                                    try {
                                        esDao.createMapping(table, map2, true);
                                    } catch (Exception e) {
                                        monitorInfo.setCount(monitorInfo.getCount() + 1);
                                        try {
                                            mr.updateErrorInfo(monitorInfo);
                                        } catch (Exception e1) {
                                            logger.error("重新表 {} 创建mapping失败,相关内容为 {},错误信息 {},mysql插入失败 {} ", table, json, e.getMessage(), e1.getMessage());
                                        }
                                        stopflag.set(0);
                                        continue;
                                    }
                                } else {
                                    esDao.createMapping(table, map2, false);
                                }

                                try {
                                    final int success;

                                    success = esDao.insert(table, map2, pkStr);
                                    if (success == 0) {
                                        ignore.decrementAndGet();
                                        number.incrementAndGet();
                                        monitorInfo.setType(4);
                                        try {
                                            mr.updateErrorInfo(monitorInfo);
                                        } catch (Exception e1) {
                                            logger.error("重新执行成功后,更改状态失败 {} ", monitorInfo.getId());
                                        }
                                    }
                                } catch (Exception e) {
                                    monitorInfo.setCount(monitorInfo.getCount() + 1);
                                    try {
                                        mr.updateErrorInfo(monitorInfo);
                                    } catch (Exception e1) {
                                        logger.error("重新{} 插入ES {} 失败,错误信息 {},mysql插入失败 {}", json, table, e.getMessage(), e1.getMessage());
                                    }
                                }
                            }
                        }
                        stopflag.set(0);
                    } catch (Exception e) {
                        logger.error("补入的数据失败 {}", e.getMessage());
                    }
                }
            }
        }){


        }.start();*/

        final List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);
        for (final KafkaStream stream : streams) {
            final String finalHostName = hostName;
            final String finalIp = ip;
            executor.submit(new Runnable() {
                public void run() {
                    final ConsumerIterator<byte[], byte[]> it = stream.iterator();
                    while (it.hasNext()) {
                        //long begin = System.currentTimeMillis();
                        stopflag.set(1);
                            final String jsonStr = new String(it.next().message());
                            if (jsonStr.trim().equals("")) {
                                stopflag.set(0);
                                continue;
                            }
                            final JSONArray packet;
                            try {
                                packet = JSONArray.parseArray(jsonStr);
                            } catch (Exception e1) {
                                error.incrementAndGet();

                                try {
                                    esDao.insertErrorInfo(finalIp, finalHostName, jsonStr, "数据转换JSONArray失败,跳过此条,错误信息 " + e1.getMessage() + "", 1, 1);
                                } catch (Exception e) {
                                    logger.error("数据  {} 转换JSONArray失败,跳过此条,错误信息 {} ,mysql插入失败 {} ", jsonStr, e1.getMessage(), e.getMessage());
                                }
                            stopflag.set(0);
                            continue;
                        }

                        for (int i = 0; i < packet.size(); i++) {
                            //long start = System.currentTimeMillis();
                            //logger.info("开始解析数据: " + start);
                            final String json = packet.getString(i);
                            final Object map12;
                            try {
                                map12 = JSONArray.parse(json);
                            } catch (Exception e) {
                                error.incrementAndGet();
                                try {
                                    esDao.insertErrorInfo(finalIp, finalHostName, json, "数据转换Object失败,跳过此条,错误信息 " + e.getMessage() + "", 1, 1);
                                } catch (Exception e1) {
                                    logger.error("数据  {} 转换Object失败,跳过此条,错误信息 {} ,mysql插入失败 {}", json, e.getMessage(), e1.getMessage());
                                }
                                stopflag.set(0);
                                continue;
                            }
                            if (!(map12 instanceof Map)) {
                                error.incrementAndGet();
                                try {
                                    esDao.insertErrorInfo(finalIp, finalHostName, json, "数据转换Object失败,跳过此条,转换结果为信息 " + map12 + "", 1, 1);
                                } catch (Exception e1) {
                                    logger.error("数据  {} 转换Object失败,跳过此条,转换结果为信息 {} ,mysql插入失败 {}", json, map12, e1.getMessage());
                                }
                                stopflag.set(0);
                                continue;
                            }

                            @SuppressWarnings("unchecked")
                            final Map<String, Object> map1 = (Map<String, Object>) map12;
                            final Object lst22 = map1.get("DATA");
                            if (!(lst22 instanceof List)) {
                                error.incrementAndGet();
                                try {
                                    esDao.insertErrorInfo(finalIp, finalHostName, json, "数据获取DATA字段失败失败,跳过此条,DATA结果为 " + lst22 + "", 1, 1);
                                } catch (Exception e1) {
                                    logger.error("数据  {} 获取DATA字段失败,跳过此条,DATA结果为 {} ,mysql插入失败 {}", json, lst22, e1.getMessage());
                                }
                                stopflag.set(0);
                                continue;
                            }
                            @SuppressWarnings("unchecked")
                            final List<Map<String, Object>> lst2 = (List<Map<String, Object>>) lst22;

                            final String table = (String) map1.get("TABLE");

                            /** 表名字段不存在*/
                            if (table == null) {
                                error.incrementAndGet();
                                try {
                                    esDao.insertErrorInfo(finalIp, finalHostName, json, "数据TABLE字段不存在,跳过此条", 1, 1);
                                } catch (Exception e1) {
                                    logger.error("数据  {} TABLE字段不存在,跳过此条 ,mysql插入失败 {}", json, e1.getMessage());
                                }
                                stopflag.set(0);
                                continue;
                            }

                            /*if (table.trim().equals("qbs_fedchubanwu")) {
                                break;
                            }*/

                            //final Integer bExsit = tableExsitMap.get(table);

                            /*if (bExsit == null) {
                                if (mr.getCache(table)) {

                                    try {
                                        esDao.createMapping(table);
                                        tableExsitMap.put(table, 1);
                                    } catch (Exception e) {
                                        try {
                                            mr.insertErrorInfo(finalIp, finalHostName, json, "es创建" + table + " mapping时失败,等待再次执行,错误信息 " + e.getMessage(), 2, 1);
                                        } catch (Exception e1) {
                                            logger.error("表 {} 创建mapping失败,相关内容为 {},错误信息 {}, mysql插入失败 {}", table, json, e.getMessage(), e1.getMessage());
                                        }
                                        ignore.incrementAndGet();

                                        stopflag.set(0);
                                        continue;
                                    }

                                } else {
                                    try {
                                        mr.insertErrorInfo(finalIp, finalHostName, json, "查询表" + table + " 时失败,等待再次执行", 2, 1);
                                    } catch (Exception e1) {
                                        logger.error("查询表 {} 失败,相关内容 {} ,mysql插入失败 {}", table, json, e1.getMessage());
                                    }
                                    ignore.incrementAndGet();

                                    stopflag.set(0);
                                    continue;
                                }
                            }*/

                            //logger.info("解析数据完成: " + (System.currentTimeMillis() - start));

                            //start = System.currentTimeMillis();
                            //logger.info("开始创建mapping: " + start);
                            /** 主键 */
                            final List<String> pkStr = mr.getPrikeyList(table);
                            for (final Map<String, Object> map2 : lst2) {
                                /** 检查表结构 */
                                if (mr.checkSchema(table, map2)) {
                                    /** 表结构有变化 */
                                    try {
                                        esDao.createMapping(table, map2, true);
                                    } catch (Exception e) {
                                        try {
                                            esDao.insertErrorInfo(finalIp, finalHostName, json, "es创建" + table + " mapping时失败,等待再次执行,错误信息 " + e.getMessage(), 2, 1);
                                        } catch (Exception e1) {
                                            logger.error("表 {} 创建mapping失败,相关内容为 {},错误信息 {},mysql插入失败 {} ", table, json, e.getMessage(), e1.getMessage());
                                        }
                                        ignore.incrementAndGet();
                                        stopflag.set(0);
                                        continue;
                                    }
                                } else {
                                    try {
                                        esDao.createMapping(table, map2, false);
                                    } catch (Exception e) {
                                        try {
                                            esDao.insertErrorInfo(finalIp, finalHostName, json, "es创建" + table + " mapping时失败,等待再次执行,错误信息 " + e.getMessage(), 2, 1);
                                        } catch (Exception e1) {
                                            logger.error("表 {} 创建mapping失败,相关内容为 {},错误信息 {},mysql插入失败 {} ", table, json, e.getMessage(), e1.getMessage());
                                        }
                                        ignore.incrementAndGet();
                                        stopflag.set(0);
                                        continue;
                                    }
                                }

                                //logger.info("创建mapping完成: " + (System.currentTimeMillis() - start));
                                //start = System.currentTimeMillis();
                                //logger.info("开始插入: " + start);
                                final int success;
                                try {

                                    success = esDao.insert(table, map2, pkStr);

                                    if (success == 0) {
                                        number.incrementAndGet();
                                    } else if (success == 1) {
                                        error.incrementAndGet();
                                        try {
                                            esDao.insertErrorInfo(finalIp, finalHostName, json, "数据无主键或主键不全", 1, 1);
                                        } catch (Exception e) {
                                            logger.error("{} 插入ES {} 失败,数据无主键或主键不全,mysql插入失败 {}", json, table, e.getMessage());
                                        }
                                    }

                                } catch (Exception e) {
                                    ignore.incrementAndGet();
                                    try {
                                        esDao.insertErrorInfo(finalIp, finalHostName, json, "插入ES " + table + " 时失败,等待再次执行,错误信息 " + e.getMessage(), 2, 1);
                                    } catch (Exception e1) {
                                        logger.error("{} 插入ES {} 失败,错误信息 {},mysql插入失败 {}", json, table, e.getMessage(), e1.getMessage());
                                    }

                                }

                                //logger.info("插入完成: " + (System.currentTimeMillis() - start));

                            }
                        }
                        //logger.info("完成: " + (System.currentTimeMillis() - begin));
                        stopflag.set(0);
                    }
                }
            });
        }

    }
}
